package avicit.bdp.dgs.metadata.handler;

import avicit.bdp.common.database.ColumnMeta;
import avicit.bdp.common.database.TableMeta;
import avicit.bdp.common.dto.MetadataStateDto;
import avicit.bdp.common.service.dto.BdpColumnDTO;
import avicit.bdp.common.service.dto.BdpTableDTO;
import avicit.bdp.common.service.dto.ColumnPartitionedDTO;
import avicit.bdp.common.utils.EntityUtils;
import avicit.bdp.common.utils.VersionUtil;
import avicit.bdp.common.utils.redis.RedisCacheHelper;
import avicit.bdp.common.utils.redis.RedisLockUtils;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.core.enums.MetadataType;
import avicit.bdp.core.enums.ModelStoreStatusEnum;
import avicit.bdp.core.enums.TableTypeEnum;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dgs.metadata.compare.pojo.CompareReporter;
import avicit.bdp.dgs.metadata.compare.pojo.MetadataTableChange;
import avicit.bdp.dgs.metadata.compare.util.DBCompareCollector;
import avicit.bdp.dgs.metadata.compare.util.MetadataCompareUtil;
import avicit.bdp.dgs.metadata.dao.MetadataBusinessDao;
import avicit.bdp.dgs.metadata.dto.MetadataBaseVersionDTO;
import avicit.bdp.dgs.metadata.dto.MetadataBaseVersionDetailDTO;
import avicit.bdp.dgs.metadata.dto.MetadataBusinessDto;
import avicit.bdp.dgs.metadata.dto.MetadataCollectTaskDTO;
import avicit.bdp.dgs.metadata.dto.MetadataCollectTaskResult;
import avicit.bdp.dgs.metadata.dto.MetadataColumnTmpDTO;
import avicit.bdp.dgs.metadata.dto.MetadataModelAttributeDTO;
import avicit.bdp.dgs.metadata.dto.MetadataTmpDTO;
import avicit.bdp.dgs.metadata.service.MetadataBaseVersionDetailService;
import avicit.bdp.dgs.metadata.service.MetadataBaseVersionService;
import avicit.bdp.dgs.metadata.service.MetadataCollectTaskResultService;
import avicit.bdp.dgs.metadata.service.MetadataCollectTaskService;
import avicit.bdp.dgs.metadata.service.MetadataColumnTmpService;
import avicit.bdp.dgs.metadata.service.MetadataModelAttributeService;
import avicit.bdp.dgs.metadata.service.MetadataStateService;
import avicit.bdp.dgs.metadata.service.MetadataTableColumnPartitionService;
import avicit.bdp.dgs.metadata.service.MetadataTableColumnService;
import avicit.bdp.dgs.metadata.service.MetadataTableService;
import avicit.bdp.dgs.metadata.service.MetadataTmpService;
import avicit.bdp.dgs.metadata.vo.MetadataMessage;
import avicit.platform6.commons.utils.ComUtil;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
import javax.annotation.Resource;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Component;

/**
 * DbmsStorageHandler: 元数据关系型数据库存储
 *
 * @author mayanj
 */
@Component
public class DbmsStorageHandler implements BaseStorageHandler {

  private final Logger logger = LoggerFactory.getLogger(DbmsStorageHandler.class);

  @Resource private MetadataCollectTaskService metadataCollectTaskService;
  @Resource private MetadataTableService metadataTableService;
  @Resource private MetadataTmpService metadataTmpService;
  @Resource private MetadataColumnTmpService metadataColumnTmpService;
  @Resource private MetadataTableColumnService metadataTableColumnService;
  @Resource private MetadataTableColumnPartitionService metadataTableColumnPartitionService;
  @Resource private MetadataCollectTaskResultService metadataCollectTaskResultService;
  @Resource private MetadataStateService metadataStateService;
  @Resource private MetadataBaseVersionService metadataBaseVersionService;
  @Resource private MetadataBaseVersionDetailService metadataBaseVersionDetailService;
  @Resource private RedisLockUtils redisLockUtils;
  @Resource private RedisCacheHelper redisCacheHelper;
  @Resource private MetadataBusinessDao metadataBusinessDao;
  @Resource private MetadataModelAttributeService metadataModelAttributeService;

  @Override
  public void handle(MetadataMessage metadataMessage) {
    Integer metadataType = metadataMessage.getMetadataType();
    String messageId = metadataMessage.getMessageId();
    if (metadataType != null
        && MetadataType.TABLE.ordinal() == metadataType
        && StringUtils.isNotBlank(messageId)) {
      String lockKey = String.format("BDP:DGS:METADATA_HANDLE_%s_%s", metadataType, messageId);
      // 上锁，保证只处理一次
      if (redisLockUtils.lock(lockKey)) {
        try {
          handleTableMetadata(metadataMessage);
        } finally {
          redisLockUtils.delete(lockKey);
        }
      }
    }
  }

  private void handleTableMetadata(MetadataMessage metadataMessage) {
    logger.debug("mysql处理表类型元数据{}", JSONUtils.toJson(metadataMessage));
    String metadataCollectTaskId = metadataMessage.getMetadataCollectTaskId();
    MetadataCollectTaskDTO metadataCollectTask =
        metadataCollectTaskService.queryById(metadataCollectTaskId);
    try {
      // 获取列信息
      Set<TableMeta> tableMetaList =
          redisCacheHelper.getCacheSetMembers(
              "MetadataCollect_" + metadataMessage.getTaskInstanceId());
      logger.info("取出redis缓存元数据{}条, 开始处理对比逻辑。", tableMetaList.size());
      if (CollectionUtils.isNotEmpty(tableMetaList)) {
        List<BdpTableDTO> newCollectList = new ArrayList<>(tableMetaList.size());
        List<MetadataStateDto> stateList = new ArrayList<>(tableMetaList.size());

        for (TableMeta tableMeta : tableMetaList) {
          if (CollectionUtils.isNotEmpty(tableMeta.getColumnMetaData())) {

            BdpTableDTO dasTableDTO = generateTableDto(metadataCollectTask, tableMeta);
            handleTableMetaColumn(dasTableDTO, tableMeta);
            newCollectList.add(dasTableDTO);

            // 技术元数据数据
            MetadataStateDto metadataState = tableMeta.getMetadataState();
            if (metadataState != null) {
              metadataState.setId(ComUtil.getId());
              metadataState.setTableName(dasTableDTO.getName());
              EntityUtils.setCreateAndUpdateInfo(metadataState);
              stateList.add(metadataState);
            }
          }
        }

        handleMetadata(metadataMessage, metadataCollectTask, newCollectList, stateList);
      }
    } catch (Exception e) {
      logger.error("元数据采集异常", e);
      throw new RuntimeException(e);
    }
  }

  /**
   * @param metadataMessage 收集待处理的元数据消息
   * @param metadataCollectTask 元数据采集任务
   * @param newTableList 采集的元数据集合
   * @param metadataStateList 技术元数据集合
   */
  private void handleMetadata(
      MetadataMessage metadataMessage,
      MetadataCollectTaskDTO metadataCollectTask,
      List<BdpTableDTO> newTableList,
      List<MetadataStateDto> metadataStateList) {

    // 历史版本中的最新版本进行对比
    List<BdpTableDTO> oldTableList =
        metadataTableService.queryTableList(
            metadataCollectTask.getDatasourceId(),
            TableTypeEnum.TABLE_METADATA.getCode(),
            true,
            null);

    List<MetadataTableChange> metadataTableChanges =
        MetadataCompareUtil.batchCompare(oldTableList, newTableList);

    CompareReporter compareReporter =
        metadataTableChanges.stream().collect(new DBCompareCollector());

    List<String> addList = compareReporter.getAddList();
    List<String> dropList = compareReporter.getDropList();
    List<String> alterList = compareReporter.getAlterList();

    logger.info("元数据对比结束，新增{}条,删除{}条，变更{}条 ", addList.size(), dropList.size(), alterList.size());

    List<String> businessCodeList = new ArrayList<>();
    businessCodeList.add("cName");
    businessCodeList.add("dataHierarchy");
    businessCodeList.add("domain");
    List<MetadataModelAttributeDTO> metadataModelAttributeList =
        metadataModelAttributeService.getListByDatasourceAndType(
            metadataCollectTask.getDatasourceType(), businessCodeList);

    for (BdpTableDTO newMetadata : newTableList) {

      // 获取逻辑模型
      BdpTableDTO logicalModelParam = new BdpTableDTO();
      logicalModelParam.setName(newMetadata.getName());
      logicalModelParam.setDataSourceId(newMetadata.getDataSourceId());
      logicalModelParam.setType(TableTypeEnum.TABLE_LOGIC.getCode());
      BdpTableDTO logicModel = metadataTableService.selectOne(logicalModelParam);

      // 获取物理模型
      BdpTableDTO physicalModelParam = new BdpTableDTO();
      physicalModelParam.setName(newMetadata.getName());
      physicalModelParam.setDataSourceId(newMetadata.getDataSourceId());
      physicalModelParam.setType(TableTypeEnum.TABLE_PHYSICAL.getCode());
      BdpTableDTO physicalModel = metadataTableService.selectOne(physicalModelParam);
      if (physicalModel != null) {
        newMetadata.setSourceTableId(physicalModel.getId());
        newMetadata.setSourceType(1);
      }

      if (addList.contains(newMetadata.getName())) {

        if (logicModel != null) {
          newMetadata.setTypeId(logicModel.getTypeId());
          newMetadata.setLayerId(logicModel.getLayerId());
        }

        String id =
            metadataTableService.insertDataTable(
                newMetadata, VersionUtil.generateVersion2Default());
        newMetadata.setId(id);

        // 处理业务元数据
        handleBusinessMetadata(
            metadataModelAttributeList, newMetadata, logicModel, newMetadata.getId());

      } else if (alterList.contains(newMetadata.getName())) {

        // 变更的元数据
        Optional<BdpTableDTO> oldTable =
            oldTableList.stream()
                .filter(r -> newMetadata.getName().equalsIgnoreCase(r.getName()))
                .findFirst();
        if (oldTable.isPresent()) {
          // 首先删除原先的记录，然后新增
          BdpTableDTO oldMetadata = oldTable.get();
          String tableId = oldMetadata.getId();

          // 从逻辑模型设置元数据的主题和数据分层
          if (logicModel != null) {
            if (StringUtils.isEmpty(oldMetadata.getTypeId())) {
              newMetadata.setTypeId(logicModel.getTypeId());
            }
            if (StringUtils.isEmpty(oldMetadata.getLayerId())) {
              newMetadata.setLayerId(logicModel.getLayerId());
            }
          }

          metadataTableService.deleteByPrimaryKey(tableId);
          metadataTableColumnService.deleteColumnByTableId(tableId);
          metadataTableColumnPartitionService.deleteByTableId(tableId);

          // 插入新的记录
          newMetadata.setId(tableId);
          String newTableVersion =
              VersionUtil.generateVersion2Next(oldTable.get().getTableVersion());
          metadataTableService.insertDataTable(newMetadata, newTableVersion);

          // 如果有变更记录，覆盖临时区数据
          MetadataTmpDTO metadataTmpDTO = metadataTmpService.queryByMetadataId(tableId);
          if (metadataTmpDTO != null) {
            metadataColumnTmpService.deleteByMetadataId(tableId);
            List<MetadataColumnTmpDTO> columnTmpDTOS = new ArrayList<>();
            for (BdpColumnDTO columnDTO : newMetadata.getColumnList()) {
              MetadataColumnTmpDTO columnTmpDTO = new MetadataColumnTmpDTO();
              BeanUtils.copyProperties(columnDTO, columnTmpDTO);
              columnTmpDTO.setMetadataId(tableId);
              if (StringUtils.isEmpty(columnTmpDTO.getId())) {
                columnTmpDTO.setId(ComUtil.getId());
              }
              columnTmpDTOS.add(columnTmpDTO);
            }
            metadataColumnTmpService.batchSave(columnTmpDTOS);
          }
        }
      }
    }

    // 处理基线版本
    handleBaseVersion(metadataCollectTask, newTableList);

    // 处理技术属性
    handleTechnologyMetaData(newTableList, metadataStateList);

    // 处理删除的元数据
    handleDeleteMetadata(metadataCollectTask, oldTableList, dropList);

    // 统计采集结果
    handleCollectResult(
        metadataMessage, metadataCollectTask, compareReporter, addList, dropList, alterList);

    logger.info("元数据任务处理完毕");
  }

  private void handleDeleteMetadata(
      MetadataCollectTaskDTO metadataCollectTask,
      List<BdpTableDTO> oldTableList,
      List<String> dropList) {
    if (CollectionUtils.isNotEmpty(dropList)) {

      List<String> dropIdList = new ArrayList<>(dropList.size());

      for (String dropTableName : dropList) {
        Optional<BdpTableDTO> oldTable =
            oldTableList.stream()
                .filter(r -> dropTableName.equalsIgnoreCase(r.getName()))
                .findFirst();
        if (oldTable.isPresent()) {
          logger.info("待删除元数据，id = {}", oldTable.get().getId());
          dropIdList.add(oldTable.get().getId());
        }
      }

      if (CollectionUtils.isNotEmpty(dropIdList)) {
        metadataTableService.batchDelete(dropIdList);
        // 从基线版本删除
        MetadataBaseVersionDTO baseVersionDTO =
            metadataBaseVersionService.getOneByDatasourceId(metadataCollectTask.getDatasourceId());
        if (baseVersionDTO != null) {
          metadataBaseVersionDetailService.batchDeleteByBaseVersionIdAndMetadataIds(
              baseVersionDTO.getId(), dropIdList);
        }
      }
    }
  }

  /**
   * 处理业务元数据
   *
   * @param metadataModelAttributeList 业务元数据元模型属性
   * @param newMetadata 元数据
   * @param logicModel 逻辑模型
   * @param metadataId 元数据ID
   */
  private void handleBusinessMetadata(
      List<MetadataModelAttributeDTO> metadataModelAttributeList,
      BdpTableDTO newMetadata,
      BdpTableDTO logicModel,
      String metadataId) {

    List<MetadataBusinessDto> metadataBusinessList = new ArrayList<>();
    for (MetadataModelAttributeDTO metadataModelAttributeDTO : metadataModelAttributeList) {
      if ("cName".equalsIgnoreCase(metadataModelAttributeDTO.getAttributeCode())) {
        MetadataBusinessDto metadataBusinessCname = new MetadataBusinessDto();
        metadataBusinessCname.setId(ComUtil.getId());
        metadataBusinessCname.setTableId(metadataId);
        metadataBusinessCname.setMetadataVersion(newMetadata.getTableVersion());
        metadataBusinessCname.setBusinessId(metadataModelAttributeDTO.getId());
        metadataBusinessCname.setValue(newMetadata.getCnName());
        EntityUtils.setCreateAndUpdateInfo(metadataBusinessCname);
        metadataBusinessList.add(metadataBusinessCname);
      } else if (logicModel != null
          && "dataHierarchy".equalsIgnoreCase(metadataModelAttributeDTO.getAttributeCode())) {
        MetadataBusinessDto metadataBusinessDataHierarchy = new MetadataBusinessDto();
        metadataBusinessDataHierarchy.setId(ComUtil.getId());
        metadataBusinessDataHierarchy.setTableId(metadataId);
        metadataBusinessDataHierarchy.setMetadataVersion(newMetadata.getTableVersion());
        metadataBusinessDataHierarchy.setBusinessId(metadataModelAttributeDTO.getId());
        metadataBusinessDataHierarchy.setValue(logicModel.getLayerId());
        EntityUtils.setCreateAndUpdateInfo(metadataBusinessDataHierarchy);
        metadataBusinessList.add(metadataBusinessDataHierarchy);
      } else if (logicModel != null
          && "domain".equalsIgnoreCase(metadataModelAttributeDTO.getAttributeCode())) {
        MetadataBusinessDto metadataBusinessDomain = new MetadataBusinessDto();
        metadataBusinessDomain.setId(ComUtil.getId());
        metadataBusinessDomain.setTableId(metadataId);
        metadataBusinessDomain.setMetadataVersion(newMetadata.getTableVersion());
        metadataBusinessDomain.setBusinessId(metadataModelAttributeDTO.getId());
        metadataBusinessDomain.setValue(logicModel.getTypeId());
        EntityUtils.setCreateAndUpdateInfo(metadataBusinessDomain);
        metadataBusinessList.add(metadataBusinessDomain);
      }
    }
    if (CollectionUtils.isNotEmpty(metadataBusinessList)) {
      metadataBusinessDao.batchInsertBusinessList(metadataBusinessList);
    }
  }

  private void handleBaseVersion(
      MetadataCollectTaskDTO metadataCollectTask, List<BdpTableDTO> newTableList) {
    // 基线版本
    // 一个数据源只有一个基线版本，基线版本不存在，自动创建，并自动将所有元数据纳入基线版本
    // 后续有变更，进行变更记录和提示（界面提示）
    MetadataBaseVersionDTO baseVersionDTO =
        metadataBaseVersionService.getOneByDatasourceId(metadataCollectTask.getDatasourceId());
    if (baseVersionDTO == null) {
      logger.info("元数据基线版本不存在，自动创建，并自动将所有元数据纳入基线版本");
      baseVersionDTO = new MetadataBaseVersionDTO();
      baseVersionDTO.setId(ComUtil.getId());
      baseVersionDTO.setDatasourceId(metadataCollectTask.getDatasourceId());
      metadataBaseVersionService.insert(baseVersionDTO);
      logger.info("元数据基线版本创建完毕");

      List<MetadataBaseVersionDetailDTO> list = new ArrayList<>();
      for (BdpTableDTO bdpTableDTO : newTableList) {
        if (StringUtils.isNotBlank(bdpTableDTO.getId())) {
          MetadataBaseVersionDetailDTO versionDetailDTO = new MetadataBaseVersionDetailDTO();
          versionDetailDTO.setId(ComUtil.getId());
          versionDetailDTO.setMetadataBaseVersionId(baseVersionDTO.getId());
          versionDetailDTO.setDatasourceId(bdpTableDTO.getDataSourceId());
          versionDetailDTO.setMetadataId(bdpTableDTO.getId());
          versionDetailDTO.setMetadataVersion(bdpTableDTO.getTableVersion());
          EntityUtils.setCreateAndUpdateInfo(versionDetailDTO);
          list.add(versionDetailDTO);
        }
      }

      if (CollectionUtils.isNotEmpty(list)) {
        logger.info("批量插入基线版本开始");
        metadataBaseVersionDetailService.batchSave(list);
        logger.info("批量插入基线版本结束");
      }

    } else {
      // 查找当前数据源下已加入基线的元数据集合
      List<String> metadataIdList =
          metadataBaseVersionDetailService.selectMetadataIdList(
              metadataCollectTask.getDatasourceId());
      if (CollectionUtils.isNotEmpty(metadataIdList)) {
        List<MetadataBaseVersionDetailDTO> list = new ArrayList<>();
        for (BdpTableDTO bdpTableDTO : newTableList) {
          // 将不在基线版本的元数据自动纳入到基线
          if (StringUtils.isNotBlank(bdpTableDTO.getId())
              && !metadataIdList.contains(bdpTableDTO.getId())) {
            logger.info("基线版本不包含的元数据ID = {}，自动纳入到基线版本中", bdpTableDTO.getId());
            MetadataBaseVersionDetailDTO versionDetailDTO = new MetadataBaseVersionDetailDTO();
            versionDetailDTO.setId(ComUtil.getId());
            versionDetailDTO.setMetadataBaseVersionId(baseVersionDTO.getId());
            versionDetailDTO.setDatasourceId(bdpTableDTO.getDataSourceId());
            versionDetailDTO.setMetadataId(bdpTableDTO.getId());
            versionDetailDTO.setMetadataVersion(bdpTableDTO.getTableVersion());
            EntityUtils.setCreateAndUpdateInfo(versionDetailDTO);
            list.add(versionDetailDTO);
          }
        }
        if (CollectionUtils.isNotEmpty(list)) {
          metadataBaseVersionDetailService.batchSave(list);
        }
      }
    }
  }

  private void handleTechnologyMetaData(
      List<BdpTableDTO> newTableList, List<MetadataStateDto> metadataStateList) {
    if (CollectionUtils.isNotEmpty(metadataStateList)) {
      Map<String, BdpTableDTO> tableMap =
          newTableList.stream().collect(Collectors.toMap(BdpTableDTO::getName, item -> item));

      for (MetadataStateDto metadataStateDto : metadataStateList) {
        if (tableMap.containsKey(metadataStateDto.getTableName())) {
          BdpTableDTO bdpTableDTO = tableMap.get(metadataStateDto.getTableName());
          metadataStateDto.setTableId(bdpTableDTO.getId());
          metadataStateDto.setTbType(bdpTableDTO.getTbType());
          metadataStateDto.setTableVersion(bdpTableDTO.getTableVersion());
        }
      }

      metadataStateService.batchInsert(metadataStateList);
      logger.info("技术元数据处理完毕");
    }
  }

  private void handleCollectResult(
      MetadataMessage metadataMessage,
      MetadataCollectTaskDTO metadataCollectTask,
      CompareReporter compareReporter,
      List<String> addList,
      List<String> dropList,
      List<String> alterList) {
    MetadataCollectTaskResult collectTaskResult = new MetadataCollectTaskResult();
    collectTaskResult.setId(ComUtil.getId());
    collectTaskResult.setMetadataCollectTaskId(metadataCollectTask.getId());
    collectTaskResult.setTaskInstanceId(metadataMessage.getTaskInstanceId());
    collectTaskResult.setTotalCount(compareReporter.getTotalCount());
    collectTaskResult.setAlterCount(compareReporter.getAlterCount());
    collectTaskResult.setAddCount(compareReporter.getAddCount());
    collectTaskResult.setDropCount(compareReporter.getDropCount());
    collectTaskResult.setAlterDetail(String.join(Constants.COMMA, alterList));
    collectTaskResult.setAddDetail(String.join(Constants.COMMA, addList));
    collectTaskResult.setDropDetail(String.join(Constants.COMMA, dropList));
    metadataCollectTaskResultService.insert(collectTaskResult);
    logger.info("元数据采集结果统计完毕");
  }

  private void handleTableMetaColumn(BdpTableDTO dasTableDTO, TableMeta tableMeta) {
    List<ColumnMeta> columnMetaList = tableMeta.getColumnMetaData();
    if (Constants.DB_TYPE_IOTDB.equals(dasTableDTO.getDataSourceType())) {
      ColumnMeta columnMeta = new ColumnMeta();
      columnMeta.setName("Time");
      columnMeta.setCnName("Time");
      columnMeta.setColumnType("timestamp");
      columnMeta.setNullable(1);
      columnMetaList.add(0, columnMeta);
    }

    List<BdpColumnDTO> columnList = new ArrayList<>();
    List<ColumnPartitionedDTO> partitionedList = new ArrayList<>();
    int order = 1;

    for (ColumnMeta columnMeta : columnMetaList) {
      if (Constants.DB_TYPE_HIVE.equals(dasTableDTO.getDataSourceType())
          && columnMeta.getPartitionable() != null
          && 1 == columnMeta.getPartitionable()) {
        partitionedList.add(createPartitionColumnByMetaData(dasTableDTO, columnMeta, order++));
      } else {
        columnList.add(createColumnByMetaData(dasTableDTO, columnMeta, order++));
      }
    }
    dasTableDTO.setPartitionedColumnList(partitionedList);
    dasTableDTO.setColumnList(columnList);
  }

  private BdpTableDTO generateTableDto(
      MetadataCollectTaskDTO metadataCollectTask, TableMeta tableMeta) {

    BdpTableDTO dasTableDTO = new BdpTableDTO();
    dasTableDTO.setName(tableMeta.getName());

    if (StringUtils.isNotBlank(tableMeta.getCnName())) {
      dasTableDTO.setCnName(tableMeta.getCnName());
    } else if (StringUtils.isNotBlank(tableMeta.getRemark())) {
      dasTableDTO.setCnName(tableMeta.getRemark());
    } else {
      dasTableDTO.setCnName(tableMeta.getName());
    }

    if (StringUtils.isNotBlank(tableMeta.getRemark())) {
      dasTableDTO.setRemark(tableMeta.getRemark());
    }

    dasTableDTO.setTypeId(metadataCollectTask.getTableBelongCatalogId());
    dasTableDTO.setType(TableTypeEnum.TABLE_METADATA.getCode());
    dasTableDTO.setDataSourceId(metadataCollectTask.getDatasourceId());
    dasTableDTO.setDataSourceType(tableMeta.getDbType());
    dasTableDTO.setMetadataCollectTaskId(metadataCollectTask.getId());
    dasTableDTO.setStoreStatus(ModelStoreStatusEnum.STORE_STATUS_SAVED.ordinal());
    dasTableDTO.setSkipAddToDb(true);
    dasTableDTO.setDataStorageFormat(tableMeta.getDataStorageFormat());
    dasTableDTO.setStorageGroup(tableMeta.getStorageGroup());
    dasTableDTO.setTbType(tableMeta.getTbType());
    return dasTableDTO;
  }

  private ColumnPartitionedDTO createPartitionColumnByMetaData(
      BdpTableDTO dasTableDTO, ColumnMeta columnMeta, int order) {
    ColumnPartitionedDTO partitionedDTO = new ColumnPartitionedDTO();
    partitionedDTO.setId(ComUtil.getId());
    if (dasTableDTO != null) {
      partitionedDTO.setTableId(dasTableDTO.getId());
    }
    partitionedDTO.setName(columnMeta.getName());
    partitionedDTO.setRemark(columnMeta.getRemark());
    partitionedDTO.setColumnType(columnMeta.getColumnType());
    partitionedDTO.setOrderBy(order);

    EntityUtils.setCreateAndUpdateInfo(partitionedDTO);
    return partitionedDTO;
  }

  private BdpColumnDTO createColumnByMetaData(
      BdpTableDTO dasTableDTO, ColumnMeta columnMeta, int order) {
    BdpColumnDTO column = new BdpColumnDTO();
    column.setId(ComUtil.getId());
    if (dasTableDTO != null) {
      column.setTableId(dasTableDTO.getId());
    }
    column.setName(columnMeta.getName());
    column.setNameOld(columnMeta.getName());
    column.setCnName(columnMeta.getCnName());
    column.setRemark(columnMeta.getRemark());
    column.setColumnLength(columnMeta.getColumnLength());
    column.setColumnPrecision(columnMeta.getColumnPrecision());
    column.setColumnType(columnMeta.getColumnType());
    column.setOrderBy(order);
    column.setModifyable(0);
    column.setNullable(columnMeta.getNullable() == null ? 0 : columnMeta.getNullable());
    column.setPrimaryKey(columnMeta.getPrimaryKey() == null ? 0 : columnMeta.getPrimaryKey());
    column.setUniqueKey(columnMeta.getUniqueKey() == null ? 0 : columnMeta.getUniqueKey());
    EntityUtils.setCreateAndUpdateInfo(column);
    return column;
  }
}
